package thread.disruptor.parallel;

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;
import thread.disruptor.generate1.Trade;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

/**
 * 并行计算
 * 在单消费者模式下，每个监听（EventHandler）要对应一个线程
 *
 * @author verlet
 * @date 2020/5/3
 */
public class Main {
    public static void main(String[] args) throws InterruptedException {
        long beginTime = System.currentTimeMillis();
        final int bufferSize = 1024;

        ExecutorService executorService = Executors.newFixedThreadPool(4);

        // 使用Executors的默认线程工厂
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        Disruptor<Trade> tradeDisruptor = new Disruptor<>(new EventFactory<Trade>() {
            @Override
            public Trade newInstance() {
                return new Trade();
            }
        }, bufferSize, threadFactory, ProducerType.SINGLE, new BusySpinWaitStrategy());

        // 串行操作
//         tradeDisruptor.handleEventsWith(new Handler1()).handleEventsWith(new Handler2()).handleEventsWith(new Handler3());

        // 并行操作
//        tradeDisruptor.handleEventsWith(new Handler1(),new Handler2(),new Handler3());

        // 菱形操作
        // 使用disruptor创建消费者组C1,C2 并行操作
//        EventHandlerGroup<Trade> tradeEventHandlerGroup = tradeDisruptor.handleEventsWith(new Handler1(), new Handler2());
        // 声明在C1,C2完事之后执行JMS消息发送操作 也就是流程走到C3
//        tradeEventHandlerGroup.then(new Handler3());

        //六边形操作 需要五个线程执行
        Handler1 h1 = new Handler1();
        Handler2 h2 = new Handler2();
        Handler3 h3 = new Handler3();
        Handler4 h4 = new Handler4();
        Handler5 h5 = new Handler5();
        tradeDisruptor.handleEventsWith(h1, h2);
        tradeDisruptor.after(h1).handleEventsWith(h4);
        tradeDisruptor.after(h2).handleEventsWith(h5);
        tradeDisruptor.after(h4, h5).handleEventsWith(h3);

        tradeDisruptor.start();
        CountDownLatch latch = new CountDownLatch(1);
        //生产者准备
        executorService.submit(new TradePublesher(latch, tradeDisruptor));
        //等待生产者
        latch.await();
        tradeDisruptor.shutdown();
        executorService.shutdown();
        System.out.println("总耗时:" + (System.currentTimeMillis() - beginTime));
    }
}
